/*
 * hashfuncs.c
 *		用于调查 HASH 索引内容的函数
 *
 * Copyright (c) 2017-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		contrib/pageinspect/hashfuncs.c
 */

#include "postgres.h"

#include "access/hash.h"
#include "access/htup_details.h"
#include "access/relation.h"
#include "catalog/pg_am.h"
#include "catalog/pg_type.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pageinspect.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/rel.h"

PG_FUNCTION_INFO_V1(hash_page_type);
PG_FUNCTION_INFO_V1(hash_page_stats);
PG_FUNCTION_INFO_V1(hash_page_items);
PG_FUNCTION_INFO_V1(hash_bitmap_info);
PG_FUNCTION_INFO_V1(hash_metapage_info);

#define IS_INDEX(r) ((r)->rd_rel->relkind == RELKIND_INDEX)
#define IS_HASH(r) ((r)->rd_rel->relam == HASH_AM_OID)

/* ------------------------------------------------
 * 单个哈希页面统计的结构
 * ------------------------------------------------
 */
typedef struct HashPageStat
{
	int			live_items;
	int			dead_items;
	int			page_size;
	int			free_size;

	/* 不透明的数据 */
	BlockNumber hasho_prevblkno;
	BlockNumber hasho_nextblkno;
	Bucket		hasho_bucket;
	uint16		hasho_flag;
	uint16		hasho_page_id;
} HashPageStat;


/*
 * 验证给定的bytea是否包含一个HASH页面，否则将报错。
 * 返回一个指向分配的、正确对齐的页面副本的指针。
 */
static Page fc_verify_hash_page(bytea *fc_raw_page, int fc_flags)
{
	Page		fc_page = get_page_from_raw(fc_raw_page);
	int			fc_pagetype = LH_UNUSED_PAGE;

	/* 将新页面视为未使用。 */
	if (!PageIsNew(fc_page))
	{
		HashPageOpaque fc_pageopaque;

		if (PageGetSpecialSize(fc_page) != MAXALIGN(sizeof(HashPageOpaqueData)))
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("input page is not a valid %s page", "hash"),
					 errdetail("Expected special size %d, got %d.",
							   (int) MAXALIGN(sizeof(HashPageOpaqueData)),
							   (int) PageGetSpecialSize(fc_page))));

		fc_pageopaque = HashPageGetOpaque(fc_page);
		if (fc_pageopaque->hasho_page_id != HASHO_PAGE_ID)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("input page is not a valid %s page", "hash"),
					 errdetail("Expected %08x, got %08x.",
							   HASHO_PAGE_ID, fc_pageopaque->hasho_page_id)));

		fc_pagetype = fc_pageopaque->hasho_flag & LH_PAGE_TYPE;
	}

	/* 检查页面类型是否合理。 */
	if (fc_pagetype != LH_OVERFLOW_PAGE && fc_pagetype != LH_BUCKET_PAGE &&
		fc_pagetype != LH_BITMAP_PAGE && fc_pagetype != LH_META_PAGE &&
		fc_pagetype != LH_UNUSED_PAGE)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid hash page type %08x", fc_pagetype)));

	/* 如果请求，验证页面类型。 */
	if (fc_flags != 0 && (fc_pagetype & fc_flags) == 0)
	{
		switch (fc_flags)
		{
			case LH_META_PAGE:
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("page is not a hash meta page")));
				break;
			case LH_BUCKET_PAGE | LH_OVERFLOW_PAGE:
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("page is not a hash bucket or overflow page")));
				break;
			case LH_OVERFLOW_PAGE:
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("page is not a hash overflow page")));
				break;
			default:
				elog(ERROR,
					 "hash page of type %08x not in mask %08x",
					 fc_pagetype, fc_flags);
				break;
		}
	}

	/*
	 * 如果是元页面，还需验证魔法数字和版本。
	 */
	if (fc_pagetype == LH_META_PAGE)
	{
		HashMetaPage fc_metap = HashPageGetMeta(fc_page);

		if (fc_metap->hashm_magic != HASH_MAGIC)
			ereport(ERROR,
					(errcode(ERRCODE_INDEX_CORRUPTED),
					 errmsg("invalid magic number for metadata"),
					 errdetail("Expected 0x%08x, got 0x%08x.",
							   HASH_MAGIC, fc_metap->hashm_magic)));

		if (fc_metap->hashm_version != HASH_VERSION)
			ereport(ERROR,
					(errcode(ERRCODE_INDEX_CORRUPTED),
					 errmsg("invalid version for metadata"),
					 errdetail("Expected %d, got %d.",
							   HASH_VERSION, fc_metap->hashm_version)));
	}

	return fc_page;
}

/* -------------------------------------------------
 * GetHashPageStatistics()
 *
 * 收集单个哈希页面的统计信息
 * -------------------------------------------------
 */
static void fc_GetHashPageStatistics(Page fc_page, HashPageStat *fc_stat)
{
	OffsetNumber fc_maxoff = PageGetMaxOffsetNumber(fc_page);
	HashPageOpaque fc_opaque = HashPageGetOpaque(fc_page);
	int			fc_off;

	fc_stat->dead_items = fc_stat->live_items = 0;
	fc_stat->page_size = PageGetPageSize(fc_page);

	/* 哈希页面不透明数据 */
	fc_stat->hasho_prevblkno = fc_opaque->hasho_prevblkno;
	fc_stat->hasho_nextblkno = fc_opaque->hasho_nextblkno;
	fc_stat->hasho_bucket = fc_opaque->hasho_bucket;
	fc_stat->hasho_flag = fc_opaque->hasho_flag;
	fc_stat->hasho_page_id = fc_opaque->hasho_page_id;

	/* 计算活动和死亡元组以及自由空间 */
	for (fc_off = FirstOffsetNumber; fc_off <= fc_maxoff; fc_off++)
	{
		ItemId		fc_id = PageGetItemId(fc_page, fc_off);

		if (!ItemIdIsDead(fc_id))
			fc_stat->live_items++;
		else
			fc_stat->dead_items++;
	}
	fc_stat->free_size = PageGetFreeSpace(fc_page);
}

/* ---------------------------------------------------
 * hash_page_type()
 *
 * 用法：SELECT hash_page_type(get_raw_page('con_hash_index', 1));
 * ---------------------------------------------------
 */
Datum hash_page_type(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	HashPageOpaque fc_opaque;
	int			fc_pagetype;
	const char *fc_type;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_page = fc_verify_hash_page(fc_raw_page, 0);

	if (PageIsNew(fc_page))
		fc_type = "unused";
	else
	{
		fc_opaque = HashPageGetOpaque(fc_page);

		/* 页面类型（标志） */
		fc_pagetype = fc_opaque->hasho_flag & LH_PAGE_TYPE;
		if (fc_pagetype == LH_META_PAGE)
			fc_type = "metapage";
		else if (fc_pagetype == LH_OVERFLOW_PAGE)
			fc_type = "overflow";
		else if (fc_pagetype == LH_BUCKET_PAGE)
			fc_type = "bucket";
		else if (fc_pagetype == LH_BITMAP_PAGE)
			fc_type = "bitmap";
		else
			fc_type = "unused";
	}

	PG_RETURN_TEXT_P(cstring_to_text(fc_type));
}

/* ---------------------------------------------------
 * hash_page_stats()
 *
 * 用法：SELECT * FROM hash_page_stats(get_raw_page('con_hash_index', 1));
 * ---------------------------------------------------
 */
Datum hash_page_stats(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	int			fc_j;
	Datum		fc_values[9];
	bool		fc_nulls[9];
	HashPageStat fc_stat;
	HeapTuple	fc_tuple;
	TupleDesc	fc_tupleDesc;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_page = fc_verify_hash_page(fc_raw_page, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);

	/* 保持编译器安静 */
	fc_stat.hasho_prevblkno = fc_stat.hasho_nextblkno = InvalidBlockNumber;
	fc_stat.hasho_flag = fc_stat.hasho_page_id = fc_stat.free_size = 0;

	fc_GetHashPageStatistics(fc_page, &fc_stat);

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupleDesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");
	fc_tupleDesc = BlessTupleDesc(fc_tupleDesc);

	MemSet(fc_nulls, 0, sizeof(fc_nulls));

	fc_j = 0;
	fc_values[fc_j++] = Int32GetDatum(fc_stat.live_items);
	fc_values[fc_j++] = Int32GetDatum(fc_stat.dead_items);
	fc_values[fc_j++] = Int32GetDatum(fc_stat.page_size);
	fc_values[fc_j++] = Int32GetDatum(fc_stat.free_size);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_stat.hasho_prevblkno);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_stat.hasho_nextblkno);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_stat.hasho_bucket);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_stat.hasho_flag);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_stat.hasho_page_id);

	fc_tuple = heap_form_tuple(fc_tupleDesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_tuple));
}

/*
 * SRF 的跨调用数据结构
 */
struct user_args
{
	Page		page;
	OffsetNumber offset;
};

/*-------------------------------------------------------
 * hash_page_items()
 *
 * 获取哈希页面中的IndexTupleData集合
 *
 * 用法：SELECT * FROM hash_page_items(get_raw_page('con_hash_index', 1));
 *-------------------------------------------------------
 */
Datum hash_page_items(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	Datum		fc_result;
	Datum		fc_values[3];
	bool		fc_nulls[3];
	uint32		fc_hashkey;
	HeapTuple	fc_tuple;
	FuncCallContext *fc_fctx;
	MemoryContext fc_mctx;
	struct user_args *fc_uargs;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	if (SRF_IS_FIRSTCALL())
	{
		TupleDesc	fc_tupleDesc;

		fc_fctx = SRF_FIRSTCALL_INIT();

		fc_mctx = MemoryContextSwitchTo(fc_fctx->multi_call_memory_ctx);

		fc_page = fc_verify_hash_page(fc_raw_page, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);

		fc_uargs = palloc(sizeof(struct user_args));

		fc_uargs->page = fc_page;

		fc_uargs->offset = FirstOffsetNumber;

		fc_fctx->max_calls = PageGetMaxOffsetNumber(fc_uargs->page);

		/* 为我们的结果类型构建一个元组描述符 */
		if (get_call_result_type(fcinfo, NULL, &fc_tupleDesc) != TYPEFUNC_COMPOSITE)
			elog(ERROR, "return type must be a row type");
		fc_tupleDesc = BlessTupleDesc(fc_tupleDesc);

		fc_fctx->attinmeta = TupleDescGetAttInMetadata(fc_tupleDesc);

		fc_fctx->user_fctx = fc_uargs;

		MemoryContextSwitchTo(fc_mctx);
	}

	fc_fctx = SRF_PERCALL_SETUP();
	fc_uargs = fc_fctx->user_fctx;

	if (fc_fctx->call_cntr < fc_fctx->max_calls)
	{
		ItemId		fc_id;
		IndexTuple	fc_itup;
		int			fc_j;

		fc_id = PageGetItemId(fc_uargs->page, fc_uargs->offset);

		if (!ItemIdIsValid(fc_id))
			elog(ERROR, "invalid ItemId");

		fc_itup = (IndexTuple) PageGetItem(fc_uargs->page, fc_id);

		MemSet(fc_nulls, 0, sizeof(fc_nulls));

		fc_j = 0;
		fc_values[fc_j++] = Int32GetDatum((int32) fc_uargs->offset);
		fc_values[fc_j++] = PointerGetDatum(&fc_itup->t_tid);

		fc_hashkey = _hash_get_indextuple_hashkey(fc_itup);
		fc_values[fc_j] = Int64GetDatum((int64) fc_hashkey);

		fc_tuple = heap_form_tuple(fc_fctx->attinmeta->tupdesc, fc_values, fc_nulls);
		fc_result = HeapTupleGetDatum(fc_tuple);

		fc_uargs->offset = fc_uargs->offset + 1;

		SRF_RETURN_NEXT(fc_fctx, fc_result);
	}

	SRF_RETURN_DONE(fc_fctx);
}

/* ------------------------------------------------
 * hash_bitmap_info()
 *
 * 获取特定溢出页面的位图信息
 *
 * 用法：SELECT * FROM hash_bitmap_info('con_hash_index'::regclass, 5);
 * ------------------------------------------------
 */
Datum hash_bitmap_info(PG_FUNCTION_ARGS)
{
	Oid			fc_indexRelid = PG_GETARG_OID(0);
	int64		fc_ovflblkno = PG_GETARG_INT64(1);
	HashMetaPage fc_metap;
	Buffer		fc_metabuf,
				fc_mapbuf;
	BlockNumber fc_bitmapblkno;
	Page		fc_mappage;
	bool		fc_bit = false;
	TupleDesc	fc_tupleDesc;
	Relation	fc_indexRel;
	uint32		fc_ovflbitno;
	int32		fc_bitmappage,
				fc_bitmapbit;
	HeapTuple	fc_tuple;
	int			fc_i,
				fc_j;
	Datum		fc_values[3];
	bool		fc_nulls[3];
	uint32	   *fc_freep;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_indexRel = relation_open(fc_indexRelid, AccessShareLock);

	if (!IS_INDEX(fc_indexRel) || !IS_HASH(fc_indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is not a %s index",
						RelationGetRelationName(fc_indexRel), "hash")));

	if (RELATION_IS_OTHER_TEMP(fc_indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary tables of other sessions")));

	if (fc_ovflblkno < 0 || fc_ovflblkno > MaxBlockNumber)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid block number")));

	if (fc_ovflblkno >= RelationGetNumberOfBlocks(fc_indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("block number %lld is out of range for relation \"%s\"",
						(long long int) fc_ovflblkno, RelationGetRelationName(fc_indexRel))));

	/* 读取元页面，以便确定使用哪个位图页面 */
	fc_metabuf = _hash_getbuf(fc_indexRel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	/*
	 * 拒绝尝试读取元页面或位图页面的位；这对
	 * 溢出页面是有意义的。
	 */
	if (fc_ovflblkno == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid overflow block number %u",
						(BlockNumber) fc_ovflblkno)));
	for (fc_i = 0; fc_i < fc_metap->hashm_nmaps; fc_i++)
		if (fc_metap->hashm_mapp[fc_i] == fc_ovflblkno)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid overflow block number %u",
							(BlockNumber) fc_ovflblkno)));

	/*
	 * 确定溢出位的编号。 这将在主桶
	 * 页面中出错，我们已经拒绝了上面的元页面和位图页面。
	 */
	fc_ovflbitno = _hash_ovflblkno_to_bitno(fc_metap, (BlockNumber) fc_ovflblkno);

	fc_bitmappage = fc_ovflbitno >> BMPG_SHIFT(fc_metap);
	fc_bitmapbit = fc_ovflbitno & BMPG_MASK(fc_metap);

	if (fc_bitmappage >= fc_metap->hashm_nmaps)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid overflow block number %u",
						(BlockNumber) fc_ovflblkno)));

	fc_bitmapblkno = fc_metap->hashm_mapp[fc_bitmappage];

	_hash_relbuf(fc_indexRel, fc_metabuf);

	/* 检查溢出页面的位图位状态 */
	fc_mapbuf = _hash_getbuf(fc_indexRel, fc_bitmapblkno, HASH_READ, LH_BITMAP_PAGE);
	fc_mappage = BufferGetPage(fc_mapbuf);
	fc_freep = HashPageGetBitmap(fc_mappage);

	fc_bit = ISSET(fc_freep, fc_bitmapbit) != 0;

	_hash_relbuf(fc_indexRel, fc_mapbuf);
	index_close(fc_indexRel, AccessShareLock);

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupleDesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");
	fc_tupleDesc = BlessTupleDesc(fc_tupleDesc);

	MemSet(fc_nulls, 0, sizeof(fc_nulls));

	fc_j = 0;
	fc_values[fc_j++] = Int64GetDatum((int64) fc_bitmapblkno);
	fc_values[fc_j++] = Int32GetDatum(fc_bitmapbit);
	fc_values[fc_j++] = BoolGetDatum(fc_bit);

	fc_tuple = heap_form_tuple(fc_tupleDesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_tuple));
}

/* ------------------------------------------------
 * hash_metapage_info()
 *
 * 获取哈希索引的元页面信息
 *
 * 用法：SELECT * FROM hash_metapage_info(get_raw_page('con_hash_index', 0))
 * ------------------------------------------------
 */
Datum hash_metapage_info(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	HashMetaPageData *fc_metad;
	TupleDesc	fc_tupleDesc;
	HeapTuple	fc_tuple;
	int			fc_i,
				fc_j;
	Datum		fc_values[16];
	bool		fc_nulls[16];
	Datum		fc_spares[HASH_MAX_SPLITPOINTS];
	Datum		fc_mapp[HASH_MAX_BITMAPS];

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_page = fc_verify_hash_page(fc_raw_page, LH_META_PAGE);

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupleDesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");
	fc_tupleDesc = BlessTupleDesc(fc_tupleDesc);

	fc_metad = HashPageGetMeta(fc_page);

	MemSet(fc_nulls, 0, sizeof(fc_nulls));

	fc_j = 0;
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_magic);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_version);
	fc_values[fc_j++] = Float8GetDatum(fc_metad->hashm_ntuples);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_metad->hashm_ffactor);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_metad->hashm_bsize);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_metad->hashm_bmsize);
	fc_values[fc_j++] = Int32GetDatum((int32) fc_metad->hashm_bmshift);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_maxbucket);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_highmask);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_lowmask);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_ovflpoint);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_firstfree);
	fc_values[fc_j++] = Int64GetDatum((int64) fc_metad->hashm_nmaps);
	fc_values[fc_j++] = ObjectIdGetDatum((Oid) fc_metad->hashm_procid);

	for (fc_i = 0; fc_i < HASH_MAX_SPLITPOINTS; fc_i++)
		fc_spares[fc_i] = Int64GetDatum((int64) fc_metad->hashm_spares[fc_i]);
	fc_values[fc_j++] = PointerGetDatum(construct_array(fc_spares,
												  HASH_MAX_SPLITPOINTS,
												  INT8OID,
												  sizeof(int64),
												  FLOAT8PASSBYVAL,
												  TYPALIGN_DOUBLE));

	for (fc_i = 0; fc_i < HASH_MAX_BITMAPS; fc_i++)
		fc_mapp[fc_i] = Int64GetDatum((int64) fc_metad->hashm_mapp[fc_i]);
	fc_values[fc_j++] = PointerGetDatum(construct_array(fc_mapp,
												  HASH_MAX_BITMAPS,
												  INT8OID,
												  sizeof(int64),
												  FLOAT8PASSBYVAL,
												  TYPALIGN_DOUBLE));

	fc_tuple = heap_form_tuple(fc_tupleDesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_tuple));
}
